package com.liuyi.netty.mqtt.processed;

import com.liuyi.netty.mqtt.common.helper.Pair;
import com.liuyi.netty.mqtt.processed.process.MqttProcess;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * Mqtt 消息处理管理器
 *
 * @author Mr.Fmy
 * @since 2021-08-19
 */
@Slf4j
public class MqttProcessManager {

    private static Map<MqttMessageType, Pair<MqttProcess, ThreadPoolExecutor>> PROCESS_MAP = new HashMap<MqttMessageType, Pair<MqttProcess, ThreadPoolExecutor>>();

    public static void putProcess(MqttMessageType type, MqttProcess process, ThreadPoolExecutor executor) {
        PROCESS_MAP.put(type, new Pair<>(process, executor));
    }

    public static void process(final ChannelHandlerContext channel, final MqttMessage msg) {
        final MqttMessageType type = msg.fixedHeader().messageType();
        if (!PROCESS_MAP.containsKey(type)) {
            throw new IllegalArgumentException("NOT type " + type.name() + " process");
        }
        final Pair<MqttProcess, ThreadPoolExecutor> pair = PROCESS_MAP.get(type);
        Runnable runnable = () -> {
            try {
                pair.getObject1().process(channel, msg);
            } catch (Exception e) {
                log.error("Mqtt Broker Process Error：{}", e.getMessage(),e);
            }
        };

        try {
            log.info("toPrepare exec");
            pair.getObject2().submit(runnable);
        } catch (RejectedExecutionException e) {
            log.error("Mqtt Broker Unable To Process Message，BeCause Thread NumberOf Biggest，Error：{}", e.getMessage());
        }
    }

}
